package tool.zyk;


import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

public class Test01_Zookeeper {

    public static void main(String[] args) {
        try {
            // 创建 ZooKeeper 对象
            ZooKeeper zooKeeper = testCreateZookeeper();
            // 在 ZooKeeper 创建数据节点
            testCreateNode(zooKeeper);
            // 在 ZooKeeper 中同步获取节点数据
            testGetdata(zooKeeper);
            // 在 ZooKeeper 中更新节点数据
            testSetdata(zooKeeper);
            // 在 ZooKeeper 异步获取节点数据
            testAsyncGetdata(zooKeeper);

            Thread.sleep(3000);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }


    private static ZooKeeper testCreateZookeeper() throws IOException, InterruptedException, KeeperException {
        final CountDownLatch countDownLatch = new CountDownLatch(1);

        // ZooKeeper 集群地址（没连接池的概念，是Session的概念）
        String connectionString = "192.168.8.51:2181,192.168.8.52:2181,192.168.8.53:2181";
        // ZooKeeper Session 数据超时时间（也就是这个Session关闭后，与这个Session相关的数据在ZooKeeper中保存多信）。
        Integer sessionTimeout = 3000;
        // ZooKeeper Session 级别 Watcher（Watch只发生在读方法上，如 get、exists）
        final ZooKeeper zooKeeper = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    Event.KeeperState state = watchedEvent.getState();
                    Event.EventType type = watchedEvent.getType();
                    String path = watchedEvent.getPath();

                    switch (state) {
                        case Unknown:
                            break;
                        case Disconnected:
                            break;
                        case NoSyncConnected:
                            break;
                        case SyncConnected:
                            countDownLatch.countDown();
                            break;
                        case AuthFailed:
                            break;
                        case ConnectedReadOnly:
                            break;
                        case SaslAuthenticated:
                            break;
                        case Expired:
                            break;
                        case Closed:
                            break;
                    }
                    switch (type) {
                        case None:
                            break;
                        case NodeCreated:
                            break;
                        case NodeDeleted:
                            break;
                        case NodeDataChanged:
                            break;
                        case NodeChildrenChanged:
                            break;
                        case DataWatchRemoved:
                            break;
                        case ChildWatchRemoved:
                            break;
                        case PersistentWatchRemoved:
                            break;
                    }

                    System.out.println("Session watch state=" + state);
                    System.out.println("Session watch type=" + type);
                    System.out.println("Session watch path=" + path);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        countDownLatch.await();
        ZooKeeper.States state = zooKeeper.getState();
        switch (state) {
            case CONNECTING:
                break;
            case ASSOCIATING:
                break;
            case CONNECTED:
                break;
            case CONNECTEDREADONLY:
                break;
            case CLOSED:
                break;
            case AUTH_FAILED:
                break;
            case NOT_CONNECTED:
                break;
        }
        System.out.println("ZooKeeper state=" + state);

        return zooKeeper;
    }

    private static void testCreateNode(ZooKeeper zooKeeper) throws IOException, InterruptedException, KeeperException {

        // 在 ZooKeeper 中创建节点
        String nodeName = zooKeeper.create("/yiqifu", "test create data".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("ZooKeeper 创建节点成功：" + nodeName);
    }

    private static void testGetdata(final ZooKeeper zooKeeper) throws IOException, InterruptedException, KeeperException {

        // 获取 ZooKeeper 节点数据，这里设置了Path级Watch
        final Stat stat = new Stat();
        byte[] nodeData = zooKeeper.getData("/yiqifu", new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    Event.KeeperState state = watchedEvent.getState();
                    Event.EventType type = watchedEvent.getType();
                    String path = watchedEvent.getPath();

                    System.out.println("Path watch state=" + state);
                    System.out.println("Path watch type=" + type);
                    System.out.println("Path watch path=" + path);

                    //zooKeeper.getData("/yiqifu", true, stat); // 这里会使用Session级Watch
                    zooKeeper.getData("/yiqifu", this, stat);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, stat);
        System.out.println("ZooKeeper 同步获取节点数据：" + new String(nodeData));
    }

    private static void testSetdata(ZooKeeper zooKeeper) throws IOException, InterruptedException, KeeperException {
        // 更新 ZooKeeper 节点数据（修改数据会触发Watch）
        zooKeeper.setData("/yiqifu", "test modify data 1 ".getBytes(), 0);
        zooKeeper.setData("/yiqifu", "test modify data 2 ".getBytes(), 1);
    }

    private static void testAsyncGetdata(ZooKeeper zooKeeper) throws IOException, InterruptedException, KeeperException {

        // 获取 ZooKeeper 节点数据（使用异步回调方式）
        Object context = new Object();
        zooKeeper.getData("/yiqifu", false, new AsyncCallback.DataCallback() {
            @Override
            public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
                System.out.println("ZooKeeper 同步获取节点数据的目录："+s);
                System.out.println("ZooKeeper 异步获取节点数据："+new String(bytes));
            }
        }, context);
    }
}

